package m3;


import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class Producer {
    public static void main(String[] args) throws MQClientException {
        //创建事务消息生产者
        TransactionMQProducer p = new TransactionMQProducer("producer-group3");
        //设置name server
        p.setNamesrvAddr("192.168.64.141:9876");

        //设置事务消息监听器
        //发送事务消息后，会触发这个监听器，执行本地事务
        //还要在这里处理事务回查操作
        p.setTransactionListener(new TransactionListener() {
            //执行本地事务的方法
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if (Math.random()<1){
                    System.out.println("由于网络中断，服务器无法获得事务状态");
                    return LocalTransactionState.UNKNOW;
                }
                System.out.println("执行本地事务");
                if (Math.random()<0.5){
                    System.out.println("本地事务执行成功");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else{
                    System.out.println("本地事务执行失败");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
//                return LocalTransactionState.COMMIT_MESSAGE;
//                return LocalTransactionState.ROLLBACK_MESSAGE;
//                return LocalTransactionState.UNKNOW;
            }
            //处理事务回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("服务器正在回查事务状态");
                if (Math.random()<0.05){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });

        //启动
        p.start();
        //发送事务消息
        while (true){
            System.out.println("输入消息");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3",s.getBytes(StandardCharsets.UTF_8));
            p.sendMessageInTransaction(msg, null);
        }

    }
}
